package com.baishancloud.log.foglog.jh

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.entity.ExternalDataEntity
import com.baishancloud.log.common.env.StreamEnv
import com.baishancloud.log.common.sink.SinkUtil
import com.baishancloud.log.common.source.{BroadcastUtil, SourceUtil}
import com.baishancloud.log.common.util.LogFormatUtil
import com.baishancloud.log.foglog._
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.datastream.BroadcastStream
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * 头条单域名数据切换到石家庄 kafka 之后的任务启动类，任务运行在金华地区集群
 *
 * @author ziqiang.wang 
 * @date 2022/2/15 10:46
 */
object FoglogDomainStatisJh extends Serializable {

  def main(args: Array[String]): Unit = {

    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    println("输入参数为：" + parameterTool.toMap.toString)

    if (!paramCheck(parameterTool)) {
      return
    }

    StreamEnv.builder(parameterTool).build()
    val foglog1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-foglog-金华").uid("7c8d6620-2918-4a5c-bc1f-77e84c8cbabd")
    val foglog2: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-foglog-喜鹊").uid("18536c2d-718c-4b8a-994b-d396bc877c8e")

    val nginx1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-nginx-金华").uid("2659d4d6-a266-4bca-a486-556510f392be")
    val nginx2: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-nginx-喜鹊").uid("e6da9ca5-7265-4bfe-9cb0-cdd268300304")


    val blblog1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-blblog-金华").uid("a6d68062-66c9-4831-ba9a-c8d93c4f594e")
    val blblog2: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-blblog-喜鹊").uid("fac2c10d-720f-4220-935a-b75f9e0d6eb4")

    val customFetch: ExternalDataEntity = ExternalDataEntity.Builder()
      .enableCountryInfoFetch()
      .enableProvinceInfoFetch()
      .enableIspInfoFetch()
      .enableNodeInfoFetch()
      .enableServiceTypeInfoFetch()
      .enableDomainInfoFetch()
      .enableSvgInfoFetch()
      .build()
    val customInfo: BroadcastStream[ExternalDataEntity] = BroadcastUtil.customInfo(customFetch)

    val fog: DataStream[(String, String)] = foglog1.union(foglog2)
      .map(("foglog", _)).name("foglogMap").uid("1f6e043a-69d-44d4-b236-3f007e5b66")

    val nginx: DataStream[(String, String)] = nginx1.union(nginx2)
      .map(("nginx", _)).name("nginxMap").uid("0c6e7ddc-fba6-58c-9b99-5fe44542f9")

    val blb: DataStream[(String, String)] = blblog1.union(blblog2).
      map(("blb", _)).name("blbMap").uid("d45f789f-f78d-4046-eb5-cf1c5b7a5d")

    fog.union(nginx).union(blb)
      .rebalance
      .map(x => (x._1, LogFormatUtil.originParse(x._2))).name("Origin").uid("652d9ece2-42feacbe-ec5354dd8f2a")
      .filter(_._2 != null).name("!=null").uid("644494f1-ded-4433-9491-5783a7f1a")
      .filter(new DomainFilterJh(parameterTool)).name("domainFilter").uid("de5be2aa6-4aa-8747-6fc6f7fce3e4")
      .connect(customInfo)
      .process(new LogParse).name("logParse").uid("f3815ce-8ada-4f9f-a3fe4825868d3")
      .keyBy(_.getKey)
      .window(TumblingProcessingTimeWindows.of(Time.minutes(parameterTool.getLong(windowSize, 1))))
      .reduce((x, y) => Log(
        x.getKey,
        LogValue(
          x.sumRequestNum + y.sumRequestNum,
          x.sumTransBytes + y.sumTransBytes,
          x.sumTransTimeMS + y.sumTransTimeMS,
          x.sumNotRedirectBytes + y.sumNotRedirectBytes,
          x.sumNotRedirectNum + y.sumNotRedirectNum,
          x.sumFirstPacketTimeMS + y.sumFirstPacketTimeMS,
          x.sumFirstFrameNum + y.sumFirstFrameNum,
          x.sumFirstFrameTimeMS + y.sumFirstFrameTimeMS,
          x.sumSslHandshakeTimeMS + y.sumSslHandshakeTimeMS,
          x.sumSslSessionReusedNum + y.sumSslSessionReusedNum,
          x.sumTransSpeedKBps + y.sumTransSpeedKBps,
          x.sumSlow256KBpsNum + y.sumSlow256KBpsNum,
          x.sumSlow512KBpsNum + y.sumSlow512KBpsNum,
          x.sumElapsedWithin1SNum + y.sumElapsedWithin1SNum,
          x.sumElapsedWithin10SNum + y.sumElapsedWithin10SNum
        ))).name("reduce").uid("7d1c9ce-d684-4886-98e5-519a5ef7")
      .map(JSONUtil.toJsonStr(_)).name("toJson").uid("ec8a1e5-6-413-9702-92d7aec9bf7")
      .addSink(SinkUtil.starRocksJsonString(parameterTool)).setParallelism(parameterTool.getInt(sinkParallel, 1)).name("starRocks").uid("a543d-d744-ba56-f7cddb72e")
    StreamEnv.execute(parameterTool)
  }

}
